package cn.xchats.rpc.client.load;

import cn.xchats.rpc.client.ClientObjectCodecHandler;
import cn.xchats.rpc.common.core.TcpFuture;
import cn.xchats.rpc.common.core.ZookeeperService;
import cn.xchats.rpc.common.dto.RequestDto;
import cn.xchats.rpc.common.type.ZookeeperNodeType;
import cn.xchats.rpc.common.util.PropertyUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/*
 *
 *@author teddy
 *@date 2018/6/27
 */
@Service
@Order(2)
public class LoadBalanceTcpClient implements CommandLineRunner {

    @Override
    public void run(String... args) throws Exception {
        initNettyClient();

        //初始化监听
        try {
            setListenterThreeThree(ZookeeperNodeType.SERVER_TYPE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private static final List<Bootstrap> bootstrapList = new ArrayList<>();
    private static final List<ChannelFuture> channelFutureList = new ArrayList<>();

    @Autowired
    private ZookeeperService zookeeperService;

    public void initNettyClient() {
        List<String> serverAddressList = LoadBalanceConfig.serverAddressList;
        if (0 == serverAddressList.size())
            throw new NullPointerException("init error : tcp server address is not find");

        //初始化所有 netty 连接
        for (int i = 0; i < serverAddressList.size(); i++) {
            Bootstrap bootstrap = new Bootstrap();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ChannelFuture channelFuture;

            bootstrap.group(workerGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.handler(new ClientObjectCodecHandler());

            try {
                String serverHost = LoadBalanceConfig.getServerHost(i);
                int serverPort = LoadBalanceConfig.getServerPort(i);

                channelFuture = bootstrap.connect(serverHost, serverPort).sync();

                //缓存
                bootstrapList.add(bootstrap);
                channelFutureList.add(channelFuture);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }


    }


    public static <T> T send(RequestDto request) {
        //负载
        int select = SelectHandler.failoverSelect(request);
        if (SelectHandler.failure == select) {
            System.out.println("服务已宕机");
            return null;
        }

        ChannelFuture channelFuture = channelFutureList.get(select);
        channelFuture.channel().writeAndFlush(request);

        TcpFuture tcpFuture = new TcpFuture(request);
        return (T) tcpFuture.getTimeout();
    }


    //3.Tree Cache
    // 监控 指定节点和节点下的所有的节点的变化--无限监听  可以进行本节点的删除(不在创建)
    private void setListenterThreeThree(String listenerPath) throws Exception {
        //设置节点的cache
        TreeCache treeCache = new TreeCache(zookeeperService.getConnection(), listenerPath);
        //设置监听器和处理过程
        treeCache.getListenable().addListener((client, event) -> {
            ChildData data = event.getData();
            if (data != null) {
                switch (event.getType()) {
                    case NODE_ADDED:
                        System.out.println("NODE_ADDED : " + data.getPath() + "  数据:" + new String(data.getData()));
                        String[] nodeSplitAdd = data.getPath().split("/");
                        //是方法节点
                        if (nodeSplitAdd.length == 4) {
                            Map<String, Map<String, String>> cacheNode = SelectHandler.cacheNode;
                            String address = nodeSplitAdd[2];
                            Map<String, String> cacheNodeValue = cacheNode.get(address);
                            //address不存在则创建
                            if (null == cacheNodeValue) {
                                cacheNodeValue = new HashMap<>();
                                cacheNode.put(nodeSplitAdd[2], cacheNodeValue);
                            }
                            //添加节点缓存
                            cacheNodeValue.put(nodeSplitAdd[3], new String(data.getData()));
                        }
                        break;
                    case NODE_REMOVED:
                        System.out.println("NODE_REMOVED : " + data.getPath() + "  数据:" + new String(data.getData()));
                        String[] nodeSplitRemove = data.getPath().split("/");
                        //是方法节点
                        if (nodeSplitRemove.length == 4) {
                            Map<String, Map<String, String>> cacheNode = SelectHandler.cacheNode;
                            String address = nodeSplitRemove[2];
                            Map<String, String> cacheNodeValue = cacheNode.get(address);
                            //address不存在则创建
                            if (null != cacheNodeValue) {
                                cacheNodeValue.remove(nodeSplitRemove[3]);
                            }
                        }
                        break;
                    case NODE_UPDATED:
                        System.out.println("NODE_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
                        // TODO
                        break;

                    default:
                        break;
                }
            } else {
                System.out.println("data is null : " + event.getType());
            }

        });
        //开始监听
        treeCache.start();

    }
}
